//********************************************************************
//	created:	10:2:2012   15:57
//	filename: 	rpcservice.cpp
//	author:		tiamo
//	purpose:	rpc service
//********************************************************************

#include "stdafx.h"
#include "rpcservice.h"

namespace rpc
{
	namespace sql
	{
		//
		// upload category
		//
		class SaveCategory : public mysql::SQLRequest
		{
		public:
			//
			// constructor
			//
			SaveCategory(mysql::Database& db, network::AsyncSocket* clientSocket) : SQLRequest(db, L"SaveCategory"), mClientSocket(clientSocket){}

		private:
			//
			// do request imp
			//
			virtual void doRequestImp();

		public:
			//
			// device name
			//
			UniString														mDeviceName;

			//
			// device token
			//
			UniString														mDeviceToken;

			//
			// dev mode
			//
			uint8_t															mDevMode;

			//
			// category filter
			//
			UniString														mCategoryFilter;

			//
			// service name
			//
			UniString														mServiceName;

			//
			// client socket
			//
			utils::SmartPointer<network::AsyncSocket>						mClientSocket;
		};
	}

	//
	// initialize
	//
	void Service::initialize()
	{
		//
		// call super
		//
		::Service::initialize();

		//
		// create listen socket
		//
		utils::Config* cfg													= Server::getSingleton()->getConfig();
		UniString address													= cfg->getString(L"/config/browser/listen_address", L"0.0.0.0");
		uint32_t port														= cfg->getInt(L"/config/browser/listen_port", 8887);
		createListenSocket(address, port);

		//
		// setup rpc method
		//
		addMethod(L"category.upload", std::bind(&Service::onUploadCategory, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
		LogInfo(L"rpc::Service(%p): running at(%ls:%u)", this, address.c_str(), port);
	}

	//
	// create listen socket
	//
	void Service::createListenSocket(UniString const& localAddress, uint32_t localPort)
	{
		utils::SmartPointer<ListenSocket> listenSocket						= new ListenSocket(this);
		listenSocket->createTcpServerSocket(localAddress, static_cast<uint16_t>(localPort), true);
		Server::getSingleton()->getSocketManager()->addSocket(listenSocket.get());
	}

	//
	// setup client socket
	//
	void Service::setupClientSocket(utils::SmartPointer<network::AsyncSocket> const& clientSocket)
	{
		clientSocket->setConnectCallback(std::bind(&Service::onSocketConnected, this, std::placeholders::_1, std::placeholders::_2));
		clientSocket->setCloseCallback(std::bind(&Service::onSocketClosed, this, std::placeholders::_1, std::placeholders::_2));
		clientSocket->setDataPacketCallback(std::bind(&Service::onSocketPacket, this, std::placeholders::_1, std::placeholders::_2));
		clientSocket->setDecodeDataCallback(std::bind(&Service::onSocketDecodePacket, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));
		Server::getSingleton()->getSocketManager()->addSocket(clientSocket);
	}

	//
	// add method
	//
	void Service::addMethod(UniString const& name, Method&& method)
	{
		mMethodList[name]													= std::move(method);
	}

	//
	// send response
	//
	void Service::sendResponse(network::AsyncSocket* theSocket, xmlrpc::Value const* response, bool isFault)
	{
		xmlrpc::Encoder encoder;
		Utf8String responseBody												= encoder.encodeResponse(response, isFault);
		uint32_t bodyLength													= static_cast<uint32_t>(responseBody.length());
		static char header[]												= "Status: 200 OK\r\nContent-Type: text/xml\r\n\r\n";
		theSocket->sendBuffer(header, sizeof(header) - sizeof(header[0]));
		theSocket->sendBuffer(responseBody.c_str(), bodyLength);
		theSocket->closeAfterSend();
	}

	//
	// send fault message
	//
	void Service::sendFaultMessage(network::AsyncSocket* theSocket, int32_t faultCode, wchar_t const* messageFormat, ...)
	{
		va_list argList;
		va_start(argList, messageFormat);
		wchar_t messageBuffer[4096]											= {0};
		platform::vsnwprintf(messageBuffer, ARRAYSIZE(messageBuffer), messageFormat, argList);
		va_end(argList);

		xmlrpc::StructValue fault;
		fault.getStruct()[L"faultCode"]										= new xmlrpc::I4Value(faultCode);
		fault.getStruct()[L"faultString"]									= new xmlrpc::StringValue(messageBuffer);
		sendResponse(theSocket, &fault, true);
	}

	//
	// save category
	//
	void Service::onSaveCategory(utils::BaseRequest* theRequest)
	{
		sql::SaveCategory* saveCategory										= static_cast<sql::SaveCategory*>(theRequest);
		LogInfo(L"rpc::Service(%p): save category filter(%ls) for device(%ls) finished(%d)", this, saveCategory->mCategoryFilter.c_str(), saveCategory->mDeviceName.c_str(), theRequest->getResult());
		xmlrpc::I8Value response(theRequest->getResult());
		sendResponse(saveCategory->mClientSocket.get(), &response);
	}

	//
	// upload category
	//
	void Service::onUploadCategory(UniString const& methodName, xmlrpc::ArrayValue const* paramList, network::AsyncSocket* theSocket)
	{
		utils::SmartPointer<sql::SaveCategory> saveCategory					= new sql::SaveCategory(mDatabase, theSocket);
		auto param															= paramList->get(0);
		saveCategory->mDeviceName											= param->getString(L"deviceName");
		saveCategory->mDeviceToken											= param->getString(L"deviceToken");
		saveCategory->mDevMode												= param->getBoolean(L"devMode") ? 1 : 0;
		saveCategory->mCategoryFilter										= param->getString(L"categoryFilter");
		saveCategory->mServiceName											= param->getString(L"serviceName");
		saveCategory->setCallback(std::bind(&Service::onSaveCategory, this, std::placeholders::_1));
		Server::getSingleton()->getSQLRequestManager()->pushPendingRequest(saveCategory.get());
	}

	//
	// socket connected
	//
	void Service::onSocketConnected(network::AsyncSocket* theSocket, bool errorState)
	{
		if(errorState)
			LogError(L"rpc::Service(%p): client socket(%ls) connect faield", this, theSocket->getAddress(false).toString(true).c_str());
		else
			LogInfo(L"rpc::Service(%p): client socket(%ls) connect succeeded", this, theSocket->getAddress(false).toString(true).c_str());
	}

	//
	// socket closed
	//
	void Service::onSocketClosed(network::AsyncSocket* theSocket, bool errorState)
	{
		if(errorState)
			LogError(L"rpc::Service(%p): client socket(%ls) closed with faield status", this, theSocket->getAddress(false).toString(true).c_str());
		else
			LogInfo(L"rpc::Service(%p): client socket(%ls) closed with succeeded status", this, theSocket->getAddress(false).toString(true).c_str());
	}

	//
	// socket decode packet
	//
	utils::SmartPointer<network::BasePacket> Service::onSocketDecodePacket(network::AsyncSocket* theSocket, void const* dataBuffer, uint16_t dataLength, uint32_t& usedLength)
	{
		//
		// decode header length
		//
		char const* charBuffer												= static_cast<char const*>(dataBuffer);
		char* headerSizeEnd													= nullptr;
		char tempBuffer[12]													= {0};
		memcpy(tempBuffer, dataBuffer, std::min<uint32_t>(sizeof(tempBuffer) - 1, dataLength));
		int headerSize														= strtol(tempBuffer, &headerSizeEnd, 0);
		
		//
		// header length should end in ':'
		//
		if(headerSizeEnd == tempBuffer || *headerSizeEnd != ':' || headerSize < 17 || headerSize > 2048)
			ThrowException(L"rpc::Service(%p): decode header length for socket (%ls) failed", this, theSocket->getAddress(false).toString(true).c_str());

		//
		// skip ':' and check the buffer contains all header content
		//
		headerSizeEnd														= const_cast<char*>(charBuffer + (headerSizeEnd - tempBuffer) + 1);
		if(dataLength < headerSize + 1)
			return nullptr;

		//
		// check content length
		//
		if(memcmp(headerSizeEnd, "CONTENT_LENGTH", 15))
			ThrowException(L"rpc::Service(%p): invalid CONENT_LENGTH magic from socket (%ls)", this, theSocket->getAddress(false).toString(true).c_str());

		//
		// decode content length
		//
		char* contentPos													= nullptr;
		memcpy(tempBuffer, headerSizeEnd + 15, std::min<uint32_t>(sizeof(tempBuffer) - 1, headerSize - 15));
		int contentSize														= strtol(tempBuffer, &contentPos, 0);

		//
		// content length should end in NULL
		//
		if(*contentPos || contentSize <= 0 || contentSize >= 2 * 1024 * 1024)
			ThrowException(L"rpc::Service(%p): invalid content length from socket (%ls)", this, theSocket->getAddress(false).toString(true).c_str());

		//
		// check body length
		//
		char const* bodyBuffer												= headerSizeEnd + headerSize + 1;
		headerSize															= static_cast<uint32_t>(bodyBuffer - charBuffer);
		uint32_t totalSize													= headerSize + contentSize;
		if(dataLength < totalSize)
			return nullptr;

		//
		// build packet
		//
		usedLength															= dataLength;
		return new RpcRequestPacket(bodyBuffer, contentSize);
	}

	//
	// socket data
	//
	void Service::onSocketPacket(network::AsyncSocket* theSocket, network::BasePacket* thePacket)
	{
		RpcRequestPacket* requestPacket										= static_cast<RpcRequestPacket*>(thePacket);
		try
		{
			xmlrpc::Decoder decoder;
			decoder.decodeRequest(std::move(requestPacket->getDataBuffer()));

			UniString const& methodName										= decoder.getMethodName();
			auto it															= mMethodList.find(methodName);
			if(it != mMethodList.end())
				it->second(methodName, decoder.getParamList(), theSocket);
			else
				sendFaultMessage(theSocket, xmlrpc::Exception::NoSuchMethodError, L"Method '%ls' not found", methodName.c_str());
		}
		catch(xmlrpc::Exception& e)
		{
			sendFaultMessage(theSocket, e.getCode(), L"%ls", e.getMessage());
		}
	}

	//
	// on network event
	//
	void Service::ListenSocket::onNetworkEvent(bool readEvent, bool writeEvent, bool errorEvent)
	{
		if(readEvent)
		{
			try
			{
				network::AddressHelper peerAddress;
				SOCKET newSocket											= accept(peerAddress);
				if(newSocket != INVALID_SOCKET)
				{
					utils::SmartPointer<network::AsyncSocket> clientSocket	= new network::AsyncSocket;
					clientSocket->attachAcceptedSocket(newSocket, peerAddress, true);
					mService->setupClientSocket(clientSocket);
				}
			}
			catch(std::exception& e)
			{
				onError(&e);
			}
		}

		if(errorEvent)
			onError(nullptr);
	}

	//
	// do request
	//
	void sql::SaveCategory::doRequestImp()
	{
		mDatabase.startTransaction();
		auto rollback														= utils::makeResourceHelper([this](){mDatabase.rollbackTransaction();});
		mysql::Statement statement											= mDatabase.createStatement();
		uint32_t id															= 0;
		statement.prepare(L"select id from subscription where device_token=? and service_name=?");
		statement.bind(L"id", false, id);
		statement.bind(L"device_token", true, mDeviceToken);
		statement.bind(L"service_name", true, mServiceName);
		statement.execute();
		bool alreadyExist													= statement.fetch();
		uint32_t nowTime													= static_cast<uint32_t>(time(nullptr));
		if(alreadyExist)
		{
			statement.prepareV(L"update subscription set device_name=?, category_filter=?, active_time=from_unixtime(%d), delete_time=0 where device_token=? and service_name=? and delete_time < from_unixtime(%d)", nowTime, nowTime);
			statement.bind(L"device_name", true, mDeviceName);
			statement.bind(L"category_filter", true, mCategoryFilter);
			statement.bind(L"device_token", true, mDeviceToken);
			statement.bind(L"service_name", true, mServiceName);
		}
		else
		{
			statement.prepareV(L"insert into subscription set device_name=?, device_token=?, dev_mode=?, category_filter=?, active_time=from_unixtime(%d), delete_time=0, service_name=?", nowTime);
			statement.bind(L"device_name", true, mDeviceName);
			statement.bind(L"device_token", true, mDeviceToken);
			statement.bind(L"dev_mode", true, mDevMode);
			statement.bind(L"category_filter", true, mCategoryFilter);
			statement.bind(L"service_name", true, mServiceName);
		}
		
		statement.execute();
		mDatabase.commitTransaction();
		rollback.dismiss();
	}
}
